package org.codehaus.activemq.store.bdb;

import com.sleepycat.je.Cursor;
import com.sleepycat.je.CursorConfig;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import javax.jms.JMSException;
import javax.transaction.xa.XAException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.message.ActiveMQXid;
import org.codehaus.activemq.service.Transaction;
import org.codehaus.activemq.service.TransactionManager;
import org.codehaus.activemq.service.impl.XATransactionCommand;
import org.codehaus.activemq.store.PreparedTransactionStore;

public class BDbPreparedTransactionStore
  implements PreparedTransactionStore
{
  private static final Log log = LogFactory.getLog(BDbMessageStore.class);
  private Database database;
  private CursorConfig cursorConfig;

  public BDbPreparedTransactionStore(Database database)
  {
    this.database = database;
  }

  public ActiveMQXid[] getXids() throws XAException {
    checkClosed();
    Cursor cursor = null;
    try {
      cursor = this.database.openCursor(BDbHelper.getTransaction(), this.cursorConfig);
      List list = new ArrayList();
      DatabaseEntry keyEntry = new DatabaseEntry();
      DatabaseEntry valueEntry = new DatabaseEntry();
      OperationStatus status = cursor.getFirst(keyEntry, valueEntry, LockMode.DEFAULT);
      while (status == OperationStatus.SUCCESS) {
        list.add(extractXid(keyEntry));
      }

      if (status != OperationStatus.NOTFOUND) {
        log.warn("Unexpected status code while recovering: " + status);
      }
      ActiveMQXid[] answer = new ActiveMQXid[list.size()];
      list.toArray(answer);
      ActiveMQXid[] arrayOfActiveMQXid1 = answer;
      return arrayOfActiveMQXid1;
    }
    catch (DatabaseException e)
    {
      ActiveMQXid[] arrayOfActiveMQXid1;
      log.error("Failed to recover prepared transaction log: " + e, e);
      throw new XAException("Failed to recover prepared transaction log. Reason: " + e);
    }
    catch (IOException e) {
      log.error("Failed to recover prepared transaction log: " + e, e);
      throw new XAException("Failed to recover prepared transaction log. Reason: " + e);
    }
    finally {
      if (cursor != null)
        try {
          cursor.close();
        }
        catch (DatabaseException e) {
          log.warn("Caught exception closing cursor: " + e, e);
        }
    }
	
  }

  public void remove(ActiveMQXid xid) throws XAException
  {
    checkClosed();
    try {
      DatabaseEntry key = new DatabaseEntry(asBytes(xid));
      OperationStatus status = this.database.delete(BDbHelper.getTransaction(), key);
      if (status != OperationStatus.SUCCESS)
        log.error("Could not delete sequenece number for: " + xid + " status: " + status);
    }
    catch (DatabaseException e)
    {
      throw new XAException("Failed to remove prepared transaction: " + xid + ". Reason: " + e);
    }
    catch (IOException e) {
      throw new XAException("Failed to remove prepared transaction: " + xid + ". Reason: " + e);
    }
  }

  public void put(ActiveMQXid xid, Transaction transaction) throws XAException {
    checkClosed();
    try {
      DatabaseEntry key = new DatabaseEntry(asBytes(xid));
      DatabaseEntry value = new DatabaseEntry(asBytes(transaction));
      this.database.put(BDbHelper.getTransaction(), key, value);
    }
    catch (Exception e) {
      throw new XAException("Failed to store prepared transaction: " + xid + ". Reason: " + e);
    }
  }

  public void loadPreparedTransactions(TransactionManager transactionManager) throws XAException {
    checkClosed();
    Cursor cursor = null;
    try {
      cursor = this.database.openCursor(BDbHelper.getTransaction(), this.cursorConfig);
      DatabaseEntry keyEntry = new DatabaseEntry();
      DatabaseEntry valueEntry = new DatabaseEntry();
      OperationStatus status = cursor.getFirst(keyEntry, valueEntry, LockMode.DEFAULT);
      while (status == OperationStatus.SUCCESS) {
        ActiveMQXid xid = extractXid(keyEntry);
        Transaction transaction = extractTransaction(valueEntry);
        transactionManager.loadTransaction(xid, transaction);
        status = cursor.getNext(keyEntry, valueEntry, LockMode.DEFAULT);
      }
      if (status != OperationStatus.NOTFOUND)
        log.warn("Unexpected status code while recovering: " + status);
    }
    catch (Exception e)
    {
      log.error("Failed to recover prepared transaction log: " + e, e);
      throw new XAException("Failed to recover prepared transaction log. Reason: " + e);
    }
    finally {
      if (cursor != null)
        try {
          cursor.close();
        }
        catch (DatabaseException e) {
          log.warn("Caught exception closing cursor: " + e, e);
        }
    }
  }

  public void start() throws JMSException
  {
  }

  public synchronized void stop() throws JMSException {
    if (this.database != null) {
      JMSException exception = BDbPersistenceAdapter.closeDatabase(this.database, null);
      this.database = null;
      if (exception != null)
        throw exception;
    }
  }

  protected ActiveMQXid extractXid(DatabaseEntry entry)
    throws IOException
  {
    return ActiveMQXid.fromBytes(entry.getData());
  }

  protected Transaction extractTransaction(DatabaseEntry entry) throws IOException, ClassNotFoundException {
    return XATransactionCommand.fromBytes(entry.getData());
  }

  private byte[] asBytes(ActiveMQXid xid) throws IOException
  {
    return xid.toBytes();
  }

  private byte[] asBytes(Transaction transaction) throws IOException, JMSException {
    if ((transaction instanceof XATransactionCommand)) {
      XATransactionCommand packetTask = (XATransactionCommand)transaction;
      return packetTask.toBytes();
    }

    throw new IOException("Unsupported transaction type: " + transaction);
  }

  protected void checkClosed() throws XAException
  {
    if (this.database == null)
      throw new XAException("Prepared Transaction Store is already closed");
  }
}